/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package cn.zzdt4j.config.springboot.starter.support;

import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.ReflectUtil;
import cn.hutool.core.util.StrUtil;
import cn.zzdt4j.common.constant.Constants;
import cn.zzdt4j.common.executor.ExecutorNotifyProperties;
import cn.zzdt4j.common.executor.ThreadPoolExecutorRegistry;
import cn.zzdt4j.common.model.excutor.ExecutorProperties;
import cn.zzdt4j.common.executor.support.BlockingQueueTypeEnum;
import cn.zzdt4j.common.executor.support.RejectedPolicyTypeEnum;
import cn.zzdt4j.common.toolkit.ThreadPoolExecutorUtil;
import cn.zzdt4j.core.config.ApplicationContextHolder;
import cn.zzdt4j.core.executor.DynamicThreadPool;
import cn.zzdt4j.core.executor.DynamicThreadPoolExecutor;
import cn.zzdt4j.core.executor.support.adpter.DynamicThreadPoolAdapterChoose;
import cn.zzdt4j.threadpool.dynamic.model.config.properties.BootstrapConfigProperties;
import cn.zzdt4j.threadpool.message.core.service.GlobalNotifyAlarmManage;
import cn.zzdt4j.threadpool.message.core.service.ThreadPoolNotifyAlarm;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;

import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Dynamic thread-pool post processor
 *
 * @author by <a href="mailto:ligang941012@gmail.com">gang.Li</a>
 * @since 2023/9/14 1:35
 */
@Slf4j
@AllArgsConstructor
public final class DynamicThreadPoolPostProcessor implements BeanPostProcessor {

    private final BootstrapConfigProperties configProperties;

    private final static int DEFAULT_ACTIVE_ALARM = 80;
    private final static int DEFAULT_CAPACITY_ALARM = 80;
    private final static int DEFAULT_INTERVAL = 5;
    private static final String DEFAULT_RECEIVES = "";

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        // Determine whether the current bean is DynamicThreadPool
        if (bean instanceof DynamicThreadPoolExecutor) {
            DynamicThreadPool dynamicThreadPool;

            // 如果当前Bean 使用了DynamicThreadPool注解使用，则获取对应的Bean 的属性
            try {
                dynamicThreadPool = ApplicationContextHolder.findAnnotationOnBean(beanName, DynamicThreadPool.class);
                if (Objects.isNull(dynamicThreadPool)) {
                    return bean;
                }
            } catch (Exception e) {
                log.error("Failed to create dynamic thread pool in annotation mode.", e);
                return bean;
            }

            ThreadPoolExecutor dynamicThreadPoolExecutor = DynamicThreadPoolAdapterChoose.unwrap(bean);
            if (ObjectUtil.isNull(dynamicThreadPoolExecutor)) {
                dynamicThreadPoolExecutor = (DynamicThreadPoolExecutor) bean;
            }

            // parse bean to dynamic thread-pool
            final ThreadPoolExecutor remoteThreadPoolExecutor = fillPoolAndRegister(((DynamicThreadPoolExecutor) dynamicThreadPoolExecutor).getThreadPoolId(), dynamicThreadPoolExecutor);
            DynamicThreadPoolAdapterChoose.replace(bean, remoteThreadPoolExecutor);
            return DynamicThreadPoolAdapterChoose.match(bean) ? bean : remoteThreadPoolExecutor;
        }
        return bean;
    }

    protected ThreadPoolExecutor fillPoolAndRegister(String threadPoolId, ThreadPoolExecutor executor) {
        ExecutorProperties executorProperties = null;
        if (configProperties.getExecutors() != null) {
            executorProperties = configProperties.getExecutors()
                    .stream()
                    .filter(each -> Objects.equals(threadPoolId, each.getThreadPoolId()))
                    .findFirst()
                    .orElseThrow(() -> new RuntimeException("The thread pool id does not exist in the configuration."));
            // 生成注册实际的动态线程池的属性
            try {
                executorProperties = buildActualExecutorProperties(executorProperties);
                // Replace thread pool properties
                threadPoolParamReplace(executor, executorProperties);
            } catch (Exception ex) {
                log.error("Failed to initialize thread pool configuration.", ex);
            }
            ThreadPoolNotifyAlarm threadPoolNotifyAlarm = buildThreadPoolNotifyAlarm(executorProperties);
            GlobalNotifyAlarmManage.put(threadPoolId, threadPoolNotifyAlarm);
        }

        // Configure the default settings
        ThreadPoolExecutorRegistry.putHolder(threadPoolId, executor,
                ObjectUtil.isNull(executorProperties) ? buildDefaultExecutorProperties(threadPoolId, executor) : executorProperties);

        return executor;
    }

    /**
     * Build thread-pool notify alarm.
     *
     * @param executorProperties executor properties
     * @return Thread-pool notify alarm
     */
    private ThreadPoolNotifyAlarm buildThreadPoolNotifyAlarm(ExecutorProperties executorProperties) {
        final ExecutorNotifyProperties notifyProperties = Optional.ofNullable(executorProperties).map(ExecutorProperties::getNotify).orElse(null);
        Assert.notNull(executorProperties);
        final Boolean isAlarm = Optional.ofNullable(executorProperties.getAlarm())
                .orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(ExecutorProperties::getAlarm).orElse(Boolean.TRUE));
        final Integer activeAlarm = Optional.ofNullable(executorProperties.getActiveAlarm())
                .orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(ExecutorProperties::getActiveAlarm).orElse(DEFAULT_ACTIVE_ALARM));
        final Integer capacityAlarm = Optional.ofNullable(executorProperties.getCapacityAlarm())
                .orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(ExecutorProperties::getCapacityAlarm).orElse(DEFAULT_CAPACITY_ALARM));
        final Integer interval = Optional.ofNullable(notifyProperties)
                .map(ExecutorNotifyProperties::getInterval)
                .orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(ExecutorProperties::getNotify).map(ExecutorNotifyProperties::getInterval).orElse(DEFAULT_INTERVAL));
        final String receives = Optional.ofNullable(notifyProperties)
                .map(ExecutorNotifyProperties::getReceives)
                .orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(ExecutorProperties::getNotify).map(ExecutorNotifyProperties::getReceives).orElse(DEFAULT_RECEIVES));

        ThreadPoolNotifyAlarm threadPoolNotifyAlarm = new ThreadPoolNotifyAlarm(isAlarm, activeAlarm, capacityAlarm);
        threadPoolNotifyAlarm.setInterval(interval);
        threadPoolNotifyAlarm.setReceives(receives);
        return threadPoolNotifyAlarm;
    }

    /**
     * Replace thread pool properties
     *
     * @param executor           old executor
     * @param executorProperties new executor properties
     */
    private void threadPoolParamReplace(ThreadPoolExecutor executor, ExecutorProperties executorProperties) {
        // setup blocking queue
        BlockingQueue workQueue = BlockingQueueTypeEnum.createBlockingQueue(executorProperties.getBlockingQueue(), executorProperties.getQueueCapacity());
        /*
         * 这里可能会报一个错误 java.lang.reflect.InaccessibleObjectException: Unable to make field private java.util.concurrent.Callable java.util.concurrent.FutureTask.callable accessible: module java.base
         * does not "opens java.util.concurrent" to unnamed module @32eebfca
         * 
         * 解决办法就是需要添加一个vm参数 在Run->Edit Configurations->Modify Options->add VM options中加 --add-opens java.base/java.util.concurrent=ALL-UNNAMED
         */
        ReflectUtil.setFieldValue(executor, "workQueue", workQueue);
        // 安全设置corePoolSize和maximumPoolSize
        ThreadPoolExecutorUtil.safeSetPoolSize(executor, executorProperties.getCorePoolSize(), executorProperties.getMaximumPoolSize());
        executor.setKeepAliveTime(executorProperties.getKeepAliveTime(), TimeUnit.SECONDS);
        executor.allowCoreThreadTimeOut(executorProperties.getAllowCoreThreadTimeOut());
        executor.setRejectedExecutionHandler(RejectedPolicyTypeEnum.createPolicy(executorProperties.getRejectedHandler()));
        if (executor instanceof DynamicThreadPoolExecutor) {
            Optional.ofNullable(executorProperties.getExecuteTimeOut())
                    .ifPresent(executorTimeout -> ((DynamicThreadPoolExecutor) executor).setExecuteTimeOut(executorTimeout));
        }
    }

    /**
     * Build actual executor properties
     *
     * @param executorProperties executor Properties
     * @return actual executor properties
     */
    private ExecutorProperties buildActualExecutorProperties(ExecutorProperties executorProperties) {
        return Optional.ofNullable(configProperties.getDefaultExecutor())
                .map(each -> buildExecutorProperties(executorProperties))
                .orElse(executorProperties);
    }

    /**
     * Build actual executor properties
     *
     * @param threadPoolId thread-pool-id
     * @param executor     dynamic thread-pool executor
     * @return executor properties
     */
    private ExecutorProperties buildDefaultExecutorProperties(String threadPoolId, ThreadPoolExecutor executor) {
        ExecutorProperties executorProperties = new ExecutorProperties();
        final BlockingQueue<Runnable> blockingQueue = executor.getQueue();
        executorProperties.setCorePoolSize(executor.getCorePoolSize())
                .setMaximumPoolSize(executor.getMaximumPoolSize())
                .setAllowCoreThreadTimeOut(executor.allowsCoreThreadTimeOut())
                .setKeepAliveTime(executor.getKeepAliveTime(TimeUnit.SECONDS))
                .setBlockingQueue(blockingQueue.getClass().getSimpleName())
                .setExecuteTimeOut(Constants.EXECUTE_TIME_OUT)
                .setQueueCapacity(blockingQueue.size() + blockingQueue.remainingCapacity())
                .setRejectedHandler(executor.getRejectedExecutionHandler().getClass().getSimpleName())
                .setThreadPoolId(threadPoolId);
        return executorProperties;
    }

    private ExecutorProperties buildExecutorProperties(ExecutorProperties executorProperties) {

        return ExecutorProperties.builder()
                .corePoolSize(Optional.ofNullable(executorProperties.getCorePoolSize())
                        .orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(ExecutorProperties::getCorePoolSize).get()))
                .maximumPoolSize(Optional.ofNullable(executorProperties.getMaximumPoolSize())
                        .orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(ExecutorProperties::getMaximumPoolSize).get()))
                .allowCoreThreadTimeOut(Optional.ofNullable(executorProperties.getAllowCoreThreadTimeOut())
                        .orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(ExecutorProperties::getAllowCoreThreadTimeOut).get()))
                .keepAliveTime(Optional.ofNullable(executorProperties.getKeepAliveTime())
                        .orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(ExecutorProperties::getKeepAliveTime).get()))
                .blockingQueue(Optional.ofNullable(executorProperties.getBlockingQueue())
                        .orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(ExecutorProperties::getBlockingQueue).get()))
                .executeTimeOut(Optional.ofNullable(executorProperties.getExecuteTimeOut())
                        .orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(ExecutorProperties::getExecuteTimeOut).get()))
                .rejectedHandler(Optional.ofNullable(executorProperties.getRejectedHandler())
                        .orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(ExecutorProperties::getRejectedHandler).get()))
                .threadPoolId(executorProperties.getThreadPoolId())
                .threadNamePrefix(StrUtil.isBlank(executorProperties.getThreadNamePrefix()) ? executorProperties.getThreadPoolId() : executorProperties.getThreadNamePrefix())
                // TODO 线程池通知类型, 容量等等扩展
                .build();
    }
}
